package com.demo.kafka.producer;

import com.demo.kafka.config.KafkaConsumeDelayMsgConfig;
import com.demo.kafka.constants.DelayTopicEnum;
import com.demo.kafka.entity.KafkaMessage;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * kafka自定义时间延迟消息服务，对应消费端{@link KafkaConsumeDelayMsgConfig}
 * @author K
 */
@Slf4j
@Service
public class KafkaDelayProducer {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 发送延时消息
     * @param topic 主题
     * @param payload 消息
     * @param delaySendTime 延时时间，毫秒级
     */
    @SneakyThrows
    public void sendDelay(String topic, Object payload, long delaySendTime) {
        if (delaySendTime < 1000) {
            kafkaTemplate.send(topic, payload);
            return;
        }
        long newDelaySendTime = delaySendTime;
        String proxyTopic = DelayTopicEnum.calculateDefault(newDelaySendTime);

        KafkaMessage kafkaMessage = new KafkaMessage();
        kafkaMessage.setDelayTime(newDelaySendTime);
        kafkaMessage.setTopic(topic);
        kafkaMessage.setPayload(payload);

        log.info("发送延时消息主题：{}, 延时时间：{}秒", proxyTopic, delaySendTime);
        kafkaTemplate.send(proxyTopic, kafkaMessage);
    }
}
